//
// Created by chnlich on 6/27/15.
//


#include "database_type.h"
#include "atomic_lib.h"

static void *ins_func(void *db);

static void ins_to_trie(struct database_node *root, struct database_item *item, int32_t key_l);

struct database_item *new_database_item(const char *key, const void *value, int32_t max_key_len,
                                        int32_t max_value_len) {
	struct database_item *ret = (struct database_item *) malloc(sizeof(struct database_item));
	int32_t i;
	ret->key = (char *) malloc(max_key_len);
	for (i = 0; i < max_key_len; i++)
		ret->key[i] = key[i];

	ret->value = malloc(max_value_len);
	for (i = 0; i < max_value_len; i++)
		*(char *) (ret->value + i) = *(char *) (value + i);
	return ret;
}

int32_t set_database_item(struct database_item *dbi, const char *key, const void *value, int32_t max_key_len,
                          int32_t max_value_len) {
	int32_t i;
	dbi->key = (char *) malloc(max_key_len);
	for (i = 0; i < max_key_len; i++)
		dbi->key[i] = key[i];

	dbi->value = malloc(max_value_len);
	for (i = 0; i < max_value_len; i++)
		*(char *) (dbi->value + i) = *(char *) (value + i);
	return 0;
}

void delete_database_item(struct database_item *dbi) {
	free(dbi->key);
	free(dbi->value);
	free(dbi);
}

void delete_database_item_list(struct database_item_list *dbil) {
	while (dbil) {
		free(dbil->dbi.key);
		free(dbil->dbi.value);
		struct database_item_list *tmp = dbil;
		dbil = dbil->next;
		free(tmp);
	}
}

struct database_node *new_database_node() {
	struct database_node *ret = (struct database_node *) malloc(sizeof(struct database_node));
	ret->c[0] = NULL;
	ret->c[1] = NULL;
	ret->v = NULL;
	return ret;
}

void delete_database_node(struct database_node *now) {
	if (now == NULL)
		return;
	delete_database_node(now->c[0]);
	delete_database_node(now->c[1]);
	if (now->v)
		delete_database_item(now->v);
	free(now);
}

int32_t database_init(struct database *db, int32_t kl, int32_t vl) {
	db->key_length = kl;
	db->value_length = vl;
	db->l = 0;
	db->r = 0;
	db->root = new_database_node();
	db->thread_signal = 1;
	assert(pthread_rwlock_init(&db->trie_lock, NULL) == 0);
	assert(pthread_rwlock_init(&db->buffer_lock, NULL) == 0);
	assert(pthread_create(&db->tid, NULL, ins_func, (void *) db) == 0);
	return 0;
}

static void *ins_func(void *_db) {
	struct database *db = (struct database *) _db;
	while (db->thread_signal) {

		if (db->l == db->r) {
			usleep(50);
			sched_yield();
			continue;
		}
		else if (pthread_rwlock_trywrlock(&db->trie_lock) == 0) {
			assert(pthread_rwlock_wrlock(&db->buffer_lock) == 0);
			if (db->buffer[db->l]->key) {
				struct database_item *tmp = new_database_item(db->buffer[db->l]->key, db->buffer[db->l]->value,
				                                              db->key_length, db->value_length);
				ins_to_trie(db->root, tmp, db->key_length);
			}
			delete_database_item(db->buffer[db->l]);
			if (++db->l == DB_BUFFER) db->l = 0;
			assert(pthread_rwlock_unlock(&db->buffer_lock) == 0);
			assert(pthread_rwlock_unlock(&db->trie_lock) == 0);
		}
	}
	return NULL;
}

static void ins_to_trie(struct database_node *root, struct database_item *item, int32_t key_l) {
	int32_t i, j;
	for (i = 0; i < key_l; i++) {
		for (j = 0; j < 8; j++) {
			int32_t dir = (item->key[i] >> (7 - j)) & 1;
			if (!root->c[dir]) {
				struct database_node *dbn = new_database_node();
				root->c[dir] = dbn;
			}
			root = root->c[dir];
		}
	}
	assert(root->v == NULL);
	root->v = item;
}

int32_t database_close(struct database *db) {
	db->thread_signal = 0;
	assert(pthread_rwlock_wrlock(&db->trie_lock) == 0);
	pthread_join(db->tid, NULL);
	assert(pthread_rwlock_wrlock(&db->buffer_lock) == 0);
	delete_database_node(db->root);
	db->l = db->r = 0;
	db->tid = 0;
	db->key_length = 0;
	db->value_length = 0;
	pthread_rwlock_destroy(&db->trie_lock);
	pthread_rwlock_destroy(&db->buffer_lock);
	return 0;
}

int32_t database_insert(struct database *db, struct database_item *dbi) {
	while (1) {
		if (FULL_BUFFER(db->l, db->r)) {
			usleep(50);
			sched_yield();
			continue;
		} else {
			pthread_rwlock_wrlock(&db->buffer_lock);
			if (FULL_BUFFER(db->l, db->r)) {
				pthread_rwlock_unlock(&db->buffer_lock);
				usleep(50);
				sched_yield();
				continue;
			}
			struct database_item *tmp = new_database_item(dbi->key, dbi->value, db->key_length, db->value_length);
			db->buffer[db->r++] = tmp;
			if (db->r == DB_BUFFER) db->r = 0;
			pthread_rwlock_unlock(&db->buffer_lock);
			break;
		}
	}
	return 0;
}

static void database_find(struct database_node *root, cond_func func, struct database_item_list **sto,
                          int32_t delete_flag) {
	if (root->v && func(root->v)) {
		if (delete_flag) {
			delete_database_item(root->v);
			root->v = NULL;
		} else {
			struct database_item_list *tmp = (struct database_item_list *) malloc(sizeof(struct database_item_list));
			tmp->dbi = *root->v;
			tmp->next = *sto;
			*sto = tmp;
		}
	}
	if (root->c[0])
		database_find(root->c[0], func, sto, delete_flag);
	if (root->c[1])
		database_find(root->c[1], func, sto, delete_flag);
}

int32_t database_delete(struct database *db, cond_func func) {
	pthread_rwlock_wrlock(&db->trie_lock);
	database_find(db->root, func, NULL, 1);
	pthread_rwlock_wrlock(&db->buffer_lock);
	int32_t i;
	for (i = db->l; i != db->r; i = (i + 1) & (DB_BUFFER - 1)) {
		if (db->buffer[i] && func(db->buffer[i])) {
			delete_database_item(db->buffer[i]);
			db->buffer[i] = 0;
		}
	}
	pthread_rwlock_unlock(&db->buffer_lock);
	pthread_rwlock_unlock(&db->trie_lock);
	return 0;
}

static void database_lrfind(struct database_node *root, struct database_keypair *dbk,
                            struct database_item_list **sto, int32_t delete_flag, int32_t current_bit,
                            int32_t left_bound, int32_t right_bound) {
	if (root->v) {
		if (delete_flag) {
			delete_database_item(root->v);
			root->v = NULL;
		} else {
			struct database_item_list *tmp = (struct database_item_list *) malloc(sizeof(struct database_item_list));
			tmp->dbi = *root->v;
			tmp->next = *sto;
			*sto = tmp;
		}
	}
	int32_t pos = (current_bit / 8);
	int32_t bit = 7 - (current_bit % 8);
	int32_t left_key = (dbk->l[pos] >> bit) & 1;
	int32_t right_key = (dbk->r[pos] >> bit) & 1;
	if (!(left_bound && left_key) && root->c[0])
		database_lrfind(root->c[0], dbk, sto, delete_flag, current_bit + 1, left_bound & (int32_t) (left_key == 0),
		                right_bound & (int32_t) (right_key == 0));
	if (!(right_bound && !right_key) && root->c[1])
		database_lrfind(root->c[1], dbk, sto, delete_flag, current_bit + 1, left_bound & (int32_t) (left_key == 1),
		                right_bound & (int32_t) (right_key == 1));

}

int32_t database_delete_from_key(struct database *db, struct database_keypair *dbk) {
	pthread_rwlock_wrlock(&db->trie_lock);
	database_lrfind(db->root, dbk, NULL, 1, 0, 1, 1);
	pthread_rwlock_wrlock(&db->buffer_lock);
	int32_t i;
	for (i = db->l; i != db->r; i = (i + 1) & (DB_BUFFER - 1)) {
		if (db->buffer[i] && key_strcmp(db->buffer[i]->key, dbk->l, db->key_length) >= 0 &&
		    key_strcmp(db->buffer[i]->key, dbk->r, db->key_length) <= 0) {
			delete_database_item(db->buffer[i]);
			db->buffer[i] = 0;
		}
	}
	pthread_rwlock_unlock(&db->buffer_lock);
	pthread_rwlock_unlock(&db->trie_lock);
	return 0;
}

int32_t database_search(struct database *db, cond_func func, struct database_item_list **dbil) {
	pthread_rwlock_rdlock(&db->trie_lock);
	struct database_item_list *tmp = NULL;
	database_find(db->root, func, &tmp, 0);
	pthread_rwlock_rdlock(&db->buffer_lock);
	int32_t i;
	for (i = db->l; i != db->r; i = (i + 1) & (DB_BUFFER - 1)) {
		if (db->buffer[i] && func(db->buffer[i])) {
			struct database_item_list *t = (struct database_item_list *) malloc(sizeof(struct database_item_list));
			t->dbi = *db->buffer[i];
			t->next = tmp;
			tmp = t;
		}
	}
	*dbil = tmp;
	pthread_rwlock_unlock(&db->buffer_lock);
	pthread_rwlock_unlock(&db->trie_lock);
	return 0;
}

int32_t database_search_from_key(struct database *db, struct database_keypair *dbk, struct database_item_list **dbil) {
	pthread_rwlock_rdlock(&db->trie_lock);
	struct database_item_list *tmp = NULL;
	database_lrfind(db->root, dbk, &tmp, 0, 0, 1, 1);
	pthread_rwlock_rdlock(&db->buffer_lock);
	int32_t i;
	for (i = db->l; i != db->r; i = (i + 1) & (DB_BUFFER - 1)) {
		if (db->buffer[i] && key_strcmp(db->buffer[i]->key, dbk->l, db->key_length) >= 0 &&
		    key_strcmp(db->buffer[i]->key, dbk->r, db->key_length) <= 0) {
			struct database_item_list *t = (struct database_item_list *) malloc(sizeof(struct database_item_list));
			t->dbi = *db->buffer[i];
			t->next = tmp;
			tmp = t;
		}
	}
	*dbil = tmp;
	pthread_rwlock_unlock(&db->buffer_lock);
	pthread_rwlock_unlock(&db->trie_lock);
	return 0;
}

static int32_t func_get_all(struct database_item *dbi) {
	(void) dbi;
	return 1;
}

int32_t database_save(struct database *db, const char *file_name) {
	// save original data
	// generate file
	pthread_rwlock_rdlock(&db->trie_lock);
	pthread_rwlock_rdlock(&db->buffer_lock);
	FILE *ftmp = fopen(file_name, "w");
	if (ftmp == NULL)
		return -1;
	// magic number
	int32_t magic = 20150705;
	fwrite(&magic, 4, 1, ftmp);
	// file start, key_length & value_length
	fwrite(&db->key_length, 4, 1, ftmp);
	fwrite(&db->value_length, 4, 1, ftmp);
	// get all data from trie
	struct database_item_list *dbil = NULL, *tmp;
	database_find(db->root, func_get_all, &dbil, 0);
	struct database_item_list *cur = dbil;
	while (cur) {
		fwrite(cur->dbi.key, 1, (uint32_t) db->key_length, ftmp);
		fwrite(cur->dbi.value, 1, (uint32_t) db->value_length, ftmp);
		tmp = cur->next;
		free(cur);
		cur = tmp;
	}

	// get all data from buffer
	int32_t i;
	for (i = db->l; i != db->r; i = (i + 1) & (DB_BUFFER - 1))
		if (db->buffer[i]) {
			fwrite(db->buffer[i]->key, 1, (uint32_t) db->key_length, ftmp);
			fwrite(db->buffer[i]->value, 1, (uint32_t) db->value_length, ftmp);
		}
	fclose(ftmp);
	pthread_rwlock_unlock(&db->buffer_lock);
	pthread_rwlock_unlock(&db->trie_lock);
	return 0;
}

int32_t database_load(struct database *db, const char *file_name) {
	// load original data
	// generate file
	int32_t r;
	FILE *ftmp = fopen(file_name, "r");
	if (ftmp == NULL)
		return -1;
	// magic number
	int32_t magic;
	r = fread(&magic, 4, 1, ftmp);
	if (r != 1 && magic != 20150705)
		return -1;

	// file start, key_length & value_length
	r = fread(&db->key_length, 4, 1, ftmp);
	if (r != 1)
		return -1;
	r = fread(&db->value_length, 4, 1, ftmp);
	if (r != 1)
		return -1;
	// insert all data into trie
	char *key_buffer = (char *) malloc(db->key_length + 1);
	char *value_buffer = (char *) malloc(db->value_length + 1);
	// generate database
	database_init(db, db->key_length, db->value_length);

	// start to insert data into database
	while (1) {
		int32_t readcnt = fread(key_buffer, 1, db->key_length, ftmp);
		if (readcnt == 0)
			break;
		if (readcnt != db->key_length)
			return -1;
		readcnt = fread(value_buffer, 1, db->value_length, ftmp);
		if (readcnt < db->value_length)
			return -1;
		struct database_item dbi;
		dbi.key = key_buffer;
		dbi.value = value_buffer;
		database_insert(db, &dbi);
	}
	free(key_buffer);
	free(value_buffer);
	fclose(ftmp);
	return 0;
}